package cn.zhang.violet.report

import cn.zhang.violet.config.ConfigHandler
import org.apache.commons.lang.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SQLContext, SaveMode}

object TerminalRadio {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("媒体分析")
    conf.setMaster("local[*]")
    //设置spark程序采用的是序列化方式
    conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)

    // 读取字典数据中的数据
    val appdict = sc.textFile(ConfigHandler.appdictFilePath)
      .map(_.split("\t", -1))
      .filter(_.length >= 5)  //过滤出数据大于5的字段
      .map(fields => (fields(4), fields(1)))  //将appid 和app那么组合一起
   //将数据收集成map
    val appdictArray = appdict.collectAsMap()
    // 广播数据(非动态)
    val apppdictBT = sc.broadcast(appdictArray)

    // 读取
    val dataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)

    dataFrame.map(str => {
      val appid = str.getAs[String]("appid")
      var appname = str.getAs[String]("appname")

      //对appname和appid进行判断
      if(StringUtils.isEmpty(appname)){
         if(StringUtils.isNotEmpty(appid)){
           //对广播数据进行判断 ，看appname是否存在，再看appid是否存在
           appname = apppdictBT.value.getOrElse(appid, appid)
         }else {
           appname ="未知"
         }

      }
      // 提取个指标相关的字段
      val reqMode = str.getAs[Int]("requestmode")
      val proNode = str.getAs[Int]("processnode")
      val effTive = str.getAs[Int]("iseffective")
      val billing = str.getAs[Int]("isbilling")
      val isbid = str.getAs[Int]("isbid")
      val iswin = str.getAs[Int]("iswin")
      val adOrderId = str.getAs[Int]("adorderid")

      val rawReq = if(reqMode == 1 && proNode >= 1) 1 else 0
      val effReq = if(reqMode == 1 && proNode >= 2) 1 else 0
      val adReq =if(reqMode == 1 && proNode == 3) 1 else 0

      val adRtbReq = if(effTive == 1 && billing == 1 && isbid == 1 && adOrderId != 0) 1 else 0
      val adSuccRtbAndCostAndConsumption = if(effTive == 1 && billing == 1 && iswin == 1){

        val winPrice = str.getAs[Double]("winprice")
        val adPayment = str.getAs[Double]("adpayment")
        (1,adPayment /1000,winPrice /1000)

      } else (0,0d,0d)
      val adShow = if(reqMode == 2 && effTive == 1) 1 else 0
      val adClick = if(reqMode == 3 && effTive == 1) 1 else 0


      //组成一个对偶元祖
      (appname,List(rawReq, effReq, adReq, adRtbReq, adSuccRtbAndCostAndConsumption._1,
        adShow, adClick, adSuccRtbAndCostAndConsumption._2, adSuccRtbAndCostAndConsumption._3))
    }).reduceByKey((a,b) =>a.zip(b).map(t => t._1 + t._2))
        .sortBy(tp => tp._2.head,false)
      .saveAsTextFile("G:\\data\\log\\app")



    sc.stop()
  }
}
